package com.example.binlog_connector_test.mysqllistener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.springframework.context.ApplicationContext;


import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


import static com.github.shyiko.mysql.binlog.event.EventType.*;


public class DataChangeListener implements BinaryLogClient.EventListener {


    private ApplicationContext applicationContext;

    DataChangeListener(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    public Map<Long, String> tableNameMap = new ConcurrentHashMap<>();


    @Override
    public void onEvent(Event event) {

        DbDataChangeEvent dbDataChangeEvent = new DbDataChangeEvent(this);

        EventType eventType = event.getHeader().getEventType();

        if (eventType == EventType.TABLE_MAP) {
            TableMapEventData tableData = event.getData();
            tableNameMap.put(tableData.getTableId(), tableData.getTable());
            return;
        }
        // 只处理添加删除更新三种操作
        if (isWrite(eventType) || isUpdate(eventType) || isDelete(eventType)) {
            if (isWrite(eventType)) {
                dbDataChangeEvent.setChangeEventTypeEnum(ChangeEventTypeEnum.EVENT_ADD);
                WriteRowsEventData data = event.getData();
                String tableName = tableNameMap.get(data.getTableId());
                tableNameMap.remove(data.getTableId());
                dbDataChangeEvent.setTableName(tableName);
                dbDataChangeEvent.setOldData(null);
                data.getRows().parallelStream().forEach(x -> {
                    dbDataChangeEvent.setNewData(Arrays.asList(x));
                    applicationContext.publishEvent(dbDataChangeEvent);
                });
                return;
            }
            if (isUpdate(eventType)) {
                dbDataChangeEvent.setChangeEventTypeEnum(ChangeEventTypeEnum.EVENT_UPDATE);
                UpdateRowsEventData data = event.getData();
                String tableName = tableNameMap.get(data.getTableId());
                tableNameMap.remove(data.getTableId());
                dbDataChangeEvent.setTableName(tableName);
                data.getRows().parallelStream().forEach(x -> {
                    dbDataChangeEvent.setOldData(Arrays.asList(x.getKey()));
                    dbDataChangeEvent.setNewData(Arrays.asList(x.getValue()));
                    applicationContext.publishEvent(dbDataChangeEvent);
                });
                return;
            }
            if (isDelete(eventType)) {
                dbDataChangeEvent.setChangeEventTypeEnum(ChangeEventTypeEnum.EVENT_DELETE);
                DeleteRowsEventData data = event.getData();
                String tableName = tableNameMap.get(data.getTableId());
                tableNameMap.remove(data.getTableId());
                dbDataChangeEvent.setTableName(tableName);
                data.getRows().parallelStream().forEach(x -> {
                    dbDataChangeEvent.setOldData(Arrays.asList(x));
                    applicationContext.publishEvent(dbDataChangeEvent);
                });
            }
        }
    }
}
